using System;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Subjects;
using Sbcx.Collections.Generic;
using StreamBase.SB.Client;
using SB = StreamBase.SB;

namespace Sbcx.ComponentModel {

    internal static class ObservableTupleFactory {

        public static IConnectableObservable<SB.Tuple> Create (StreamBaseClient streamBaseClient, Func<StreamBaseClient, SB.StreamProperties> subscription, ObservableOptions options, ISubject<SB.Tuple> subject) {
            return new ConnectedSubject<SB.Tuple> (reset =>
                SubscribeToStream (reset, streamBaseClient, subscription, options, subject), subject);
        }

        static IDisposable SubscribeToStream (Action reset, StreamBaseClient streamBaseClient, Func<StreamBaseClient, SB.StreamProperties> subscription, ObservableOptions options, ISubject<SB.Tuple> subject) {
            bool scheduleImmediately = null == options.PollInterval;
            Action worker = null;

            // subscribe to stream
            var streamProperties = subscription (streamBaseClient);

            // create schedule Action based on options
            var schedule = scheduleImmediately ?
                            (Action<Action>)(action => 
                                options.DueTimeScheduler.Schedule (action)) :
                            (Action<Action>)(action =>
                                options.DueTimeScheduler.Schedule (options.PollInterval.Value, action));

            // worker will run forever (if duetime is not set), or at the interval
            worker = scheduleImmediately ?
                (Action)(() => {
                    while (TryDequeue (streamBaseClient, subject,options))
                        ;
                }) :
                (Action)(() => {
                    if (TryDequeue (streamBaseClient, subject, options))
                        schedule (worker);
                });

            schedule (worker);

            // when connection is disposed, it will unsubscribe
            return Disposable.Create (() => {
                try {
                    streamBaseClient.Unsubscribe (streamProperties);
                }
                catch (Exception ex) {
                    subject.OnError (ex);
                }
                finally {
                    reset ();
                }
            });
        }

        static bool TryDequeue (StreamBaseClient streamBaseClient, ISubject<SB.Tuple> subject, ObservableOptions options) {
            Exception exception;
            try {
                DequeueResult result;
                result = streamBaseClient.Dequeue (options.TimeoutMS);

                if (DequeueResult.GOOD == result.Status) {
                    int tupleCount = result.GetTupleCount ();
                    for (int i = 0; i < tupleCount; i++)
                        subject.OnNext (result[i]);
                    exception = null;
                }
                else if (DequeueResult.TIMEOUT == result.Status)
                    exception = null; // no results
                else if (DequeueResult.CLOSED == result.Status) {
                    exception = new SB.StreamBaseException ("CLOSED");
                }
                else if (DequeueResult.UNINITIALIZED == result.Status) {
                    exception = new SB.StreamBaseException ("UNINITIALIZED");
                }
                else
                    exception = new NotSupportedException (string.Concat ("DequeueResult.Status ", result.Status));
            }
            catch (Exception ex) {
                exception = ex;
            }

            bool success = null == exception;
            if (!success)
                subject.OnError (exception);

            return success;
        }
    }
}